package uefs.springboot.rabbitmq.a0test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Delivery;
import com.rabbitmq.client.Envelope;

/**
 * @category rabbitmq纯java代码实例
 */
public class RabbitmqExample {
	public static void main(String[] args) throws Exception {
		//RabbitmqExample.workQueueProducer();
		//RabbitmqExample.workQueueConsumer();

		//RabbitmqExample.workQueueProducer2();
		//RabbitmqExample.workQueueConsumer21();
		//RabbitmqExample.workQueueConsumer22();
		//
		fanoutSender();
	}

	/**
	 * @category 工作队列的生产者
	 */
	private static void workQueueProducer() throws IOException, TimeoutException {
		//2.创建连接
		Connection connection = connectionGet();
		//3.创建频道
		Channel channel = connection.createChannel();

		//4.通道声明队列
		/**
		 * 声明队列，如果Rabbit中没有此队列将自动创建
		 * 参数1:队列名称
		 * 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
		 * 参数3:是否独占本连接
		 * 参数4:是否在不使用的时候队列自动删除
		 * 参数5:其他参数
		 * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
		 */
		channel.queueDeclare(WORK_QUEUE, true, false, false, null);
		//5.发送消息
		for (int i = 0; i < 30; i++) {
			//5、制定消息
			String message = "hello world to rabbitmq for work queues" + i;
			/*
			 * 消息发布方法
			 * 参数1:交换机Exchange名称:如果没有指定空字符串(表示使用默认的交换机)使用Default Exchange
			 * 参数2:路由key-routingKey,简单模式中可以使用队列名称.是用于Exchange（交换机）将消息转发到指定的消息队列
			 * 参数3:消息其他属性
			 * 参数4:消息内容,消息体
			 * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
			 * 默认的交换机，routingKey等于队列名称
			 */
			//6、发送消息，使用默认交换机
			channel.basicPublish("", WORK_QUEUE, null, message.getBytes());
			System.out.println("消息已发送:" + message);
		}
		//关闭资源
		resourceClose(channel);
		resourceClose(connection);
	}

	/**
	 * @category 工作队列的消费者
	 */
	private static void workQueueConsumer() throws IOException, TimeoutException {
		Connection connection = connectionGet();
		Channel channel = connection.createChannel();
		//通道绑定队列
		channel.queueDeclare(WORK_QUEUE, true, false, false, null);//通道绑定邮件队列

		//String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
		DefaultConsumer consumer = new DefaultConsumer(channel) {
			/**
			  * 消费者接收消息调用此方法
			  * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
			  * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
			    (收到消息失败后是否需要重新发送)
			  * @param properties
			  * @param body
			  * @throws IOException
			 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
			 */
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				//交换机
				String exchange = envelope.getExchange();
				//路由key
				String routingKey = envelope.getRoutingKey();
				envelope.getDeliveryTag();
				String msg = new String(body, "utf-8");
				System.out.println("mq收到的消息是：" + msg);
			}

		};
		System.out.println("消费者启动成功！");
		channel.basicConsume(WORK_QUEUE, true, consumer);

		resourceClose(channel);
		resourceClose(connection);
	}

	private static void workQueueProducer2() throws Exception {
		Connection connection = connectionGet();
		Channel channel = connection.createChannel();
		channel.queueDeclare("work", false, false, false, null);

		for (int i = 0; i < 100; i++) {
			channel.basicPublish("", "work", null, ("工作模式发送的第 (" + i + ") 个消息").getBytes());
		}
		resourceClose(channel);
		resourceClose(connection);
	}

	private static void workQueueConsumer21() throws Exception {
		Connection conn = connectionGet();
		Channel channel = conn.createChannel();
		channel.queueDeclare("work", false, false, false, null);

		channel.basicQos(1);
		DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				//交换机
				String exchange = envelope.getExchange();
				//路由key
				String routingKey = envelope.getRoutingKey();
				envelope.getDeliveryTag();
				String msg = new String(body, "utf-8");
				System.out.println("workQueueConsumer21 msg" + msg);
				try {
					Thread.sleep(50);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				//手动发送回执
				channel.basicAck(envelope.getDeliveryTag(), false);
			}
		};
		//其中第二参数表示消费者接收消息后是否自动返回回执
		channel.basicConsume("work", false, consumer);
	}

	private static void workQueueConsumer22() throws Exception {
		Connection conn = connectionGet();
		Channel channel = conn.createChannel();
		channel.queueDeclare("work", false, false, false, null);

		channel.basicQos(1);
		//DefaultConsumer consumer = new DefaultConsumer(channel);
		DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
					byte[] body) throws IOException {
				//交换机
				String exchange = envelope.getExchange();
				//路由key
				String routingKey = envelope.getRoutingKey();
				envelope.getDeliveryTag();
				String msg = new String(body, "utf-8");
				System.out.println("workQueueConsumer22 msg" + msg);
				try {
					Thread.sleep(50);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				//手动发送回执
				channel.basicAck(envelope.getDeliveryTag(), false);
			}
		};
		//其中第二参数表示消费者接收消息后是否自动返回回执
		channel.basicConsume("work", false, consumer);
	}

	private static void fanoutSender() throws Exception {
		Connection connection = connectionGet();
		Channel channel = connection.createChannel();
		// 创建交换机
		/*
		 *  参数：
		 *      Exchange: 自定义交换机名称，接受端声明交换机的名字需和它保持一致
		 *      type:     交换机类型，取值范围(fanout(订阅/发布),direct(路由模式),topic(主题))
		 */
		channel.exchangeDeclare("fanoutEx", "fanout");
		// 发送消息
		for (int i = 0; i < 100; i++) {
			channel.basicPublish("fanoutEx", "", null, ("订阅/发布模式发送的第 （" + i + "） 个消息").getBytes());
		}
		resourceClose(channel);
		resourceClose(connection);
	}

	// 工作模式，队列名称 。多个消费端消费同一个队列中的消息，队列采用轮询的方式将消息是平均发送给消费者；
	public static final String WORK_QUEUE = "work_queue";

	public static Connection connectionGet() throws IOException, TimeoutException {
		//1.创建连接工厂(设置RabbitMq的连接参数)
		ConnectionFactory connectionFactory = new ConnectionFactory();
		//设置主机:默认:localhost
		connectionFactory.setHost("192.168.31.65");
		//连接端口:默认:5672
		connectionFactory.setPort(5672);
		//虚拟主机:默认:/
		connectionFactory.setVirtualHost("/");
		//用户名:默认:guest
		connectionFactory.setUsername("admin");
		//密码:默认:guest
		connectionFactory.setPassword("123456");
		//2.创建连接
		Connection connection = connectionFactory.newConnection();
		//返回
		return connection;
	}

	public static void resourceClose(Channel channel) {
		try {
			channel.close();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (TimeoutException e) {
			e.printStackTrace();
		}
	}

	public static void resourceClose(Connection connection) {
		try {
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
